package com.zz21

import java.io.{File, PrintWriter}

import com.gosun.{DataEncry, getResult}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK

import scala.collection.JavaConversions.asScalaBuffer
import scala.collection.mutable.ArrayBuffer

object zz21_vw_houseinfo {
  //Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession
      .builder()
      .config("spark.network.timeout", "1200")
      .config("spark.kryoserializer.buffer.max","2000")
      .config("spark.executor.memory","6g")
      //.master("local[*]")
      .appName("SQLContextApp")
      .getOrCreate()
    val options = Map(
      "es.nodes.wan.only" -> "true",
      "es.nodes" -> "192.168.130.87",
      "es.port" -> "9200",
      "es.read.field.as.array.include" -> "arr1, arr2",
      "es.scroll.size" -> "10000",
      "es.input.use.sliced.partitions" -> "false"
    )
    val index = "zz21_vw_houseinfo"
    val frame: DataFrame = sparkSession
      .read
      .format("es")
      .options(options)
      .load(index)
      .persist(StorageLevel.MEMORY_AND_DISK)


    val resSchema = StructType(
      List(
        StructField("LANDDOCUMENTS", StringType, true),
        StructField("HOUSERIGHTNUMBERDATE", StringType, true),
        StructField("HOUSERIGHTNUMBER", StringType, true),
        StructField("HOUSINGVOUCHERS", StringType, true),
        StructField("RENTALBUILDINGS", StringType, true),
        StructField("OWNPROPERTY", StringType, true),
        StructField("HOUSESOURCE", StringType, true),
        StructField("HOUSEUSES", StringType, true),
        StructField("HOUSEAREA", StringType, true),
        StructField("HOUSESTRUCTURE", StringType, true),
        StructField("BUILTYEAR", StringType, true),
        StructField("HOUSEDOORMODEL", StringType, true),
        StructField("MOBILENUMBER", StringType, true),
        StructField("TELEPHONE", StringType, true),
        StructField("HOUSEOWNERIDCARDNO", StringType, true),
        StructField("HOUSEOWNER", StringType, true),
        StructField("PROPERTYNAME", StringType, true),
        StructField("BUILDINGUSES", StringType, true),
        StructField("BUILDINGNAME", StringType, true),
        StructField("ISRENTALHOUSE", StringType, true),
        StructField("ADDRESS", StringType, true),
        StructField("ROOM", StringType, true),
        StructField("UNIT", StringType, true),
        StructField("BLOCK", StringType, true),
        StructField("COMMUNITY", StringType, true),
        StructField("ADDRESSCODE", StringType, true),
        StructField("ADDRESSTYPE", StringType, true),
        StructField("HOUSECODE", StringType, true),
        StructField("BUILDINGID", StringType, true),
        StructField("ORGINTERNALCODE", StringType, true),
        StructField("ORGID", StringType, true),
        StructField("ID", StringType, true),
        StructField("CENTERLAT2", StringType, true),
        StructField("CENTERLON2", StringType, true),
        StructField("NOWLIVEADDRESS", StringType, true),
        StructField("SOURCESSTATE", StringType, true),
        StructField("FEATUREID", StringType, true),
        StructField("CENTERLAT", StringType, true),
        StructField("CENTERLON", StringType, true),
        StructField("BUILDDATASID", StringType, true),
        StructField("UPDATEDATE", StringType, true),
        StructField("CREATEDATE", StringType, true),
        StructField("UPDATEUSER", StringType, true),
        StructField("CREATEUSER", StringType, true),
        StructField("CENTERY", StringType, true),
        StructField("CENTERX", StringType, true),
        StructField("REMARK", StringType, true),
        StructField("IMGURL", StringType, true),
        StructField("PROPERTYPERSONMOBILE", StringType, true),
        StructField("PROPERTYPERSONTEL", StringType, true),
        StructField("CERTIFICATENUMBE", StringType, true),
        StructField("CERTIFICATETYPE", StringType, true),
        StructField("SIMPLEPINYIN", StringType, true),
        StructField("FULLPINYIN", StringType, true),
        StructField("NAME", StringType, true),
        StructField("PROPERTYTYPES", StringType, true),
        StructField("LANDRIGHTSNUMBEDATE", StringType, true),
        StructField("LANDRIGHTSNUMBE", StringType, true),
        StructField("MOBILENUMBER_AES", StringType, true),
        StructField("TELEPHONE_AES", StringType, true),
        StructField("HOUSEOWNERIDCARDNO_AES", StringType, true),
        StructField("CERTIFICATENUMBE_AES", StringType, true)

      )
    )
    val value: RDD[Row] = frame.rdd.mapPartitions(iter => {
      val list = ArrayBuffer[Row]()
      while (iter.hasNext) {
        val row: Row = iter.next()
        val str1 = row.getAs[String]("LANDDOCUMENTS")
        val str2 = row.getAs[String]("HOUSERIGHTNUMBERDATE")
        val str3 = row.getAs[String]("HOUSERIGHTNUMBER")
        val str4 = row.getAs[String]("HOUSINGVOUCHERS")
        val str5 = row.getAs[String]("RENTALBUILDINGS")
        val str6 = row.getAs[String]("OWNPROPERTY")
        val str7 = row.getAs[String]("HOUSESOURCE")
        val str8 = row.getAs[String]("HOUSEUSES")
        val str9 = row.getAs[String]("HOUSEAREA")
        val str10 = row.getAs[String]("HOUSESTRUCTURE")
        val str11 = row.getAs[String]("BUILTYEAR")
        val str12 = row.getAs[String]("HOUSEDOORMODEL")
        val str13 = DataEncry.changPhone(row.getAs[String]("MOBILENUMBER"))
        val str14 = DataEncry.changPhone(row.getAs[String]("TELEPHONE"))
        val str15 = DataEncry.changIDcard(row.getAs[String]("HOUSEOWNERIDCARDNO"))
        val str16 = row.getAs[String]("HOUSEOWNER")
        val str17 = row.getAs[String]("PROPERTYNAME")
        val str18 = row.getAs[String]("BUILDINGUSES")
        val str19 = row.getAs[String]("BUILDINGNAME")
        val str20 = row.getAs[String]("ISRENTALHOUSE")
        val str21 = DataEncry.changAddress(row.getAs[String]("ADDRESS"))
        val str22 = row.getAs[String]("ROOM")
        val str23 = row.getAs[String]("UNIT")
        val str24 = row.getAs[String]("BLOCK")
        val str25 = row.getAs[String]("COMMUNITY")
        val str26 = row.getAs[String]("ADDRESSCODE")
        val str27 = row.getAs[String]("ADDRESSTYPE")
        val str28 = row.getAs[String]("HOUSECODE")
        val str29 = row.getAs[String]("BUILDINGID")
        val str30 = row.getAs[String]("ORGINTERNALCODE")
        val str31 = row.getAs[String]("ORGID")
        val str32 = row.getAs[String]("ID")
        val str33 = row.getAs[String]("CENTERLAT2")
        val str34 = row.getAs[String]("CENTERLON2")
        val str35 = DataEncry.changAddress(row.getAs[String]("NOWLIVEADDRESS"))
        val str36 = row.getAs[String]("SOURCESSTATE")
        val str37 = row.getAs[String]("FEATUREID")
        val str38 = row.getAs[String]("CENTERLAT")
        val str39 = row.getAs[String]("CENTERLON")
        val str40 = row.getAs[String]("BUILDDATASID")
        val str41 = row.getAs[String]("UPDATEDATE")
        val str42 = row.getAs[String]("CREATEDATE")
        val str43 = DataEncry.changName(row.getAs[String]("UPDATEUSER"))
        val str44 = DataEncry.changName(row.getAs[String]("CREATEUSER"))
        val str45 = row.getAs[String]("CENTERY")
        val str46 = row.getAs[String]("CENTERX")
        val str47 = row.getAs[String]("REMARK")
        val str48 = row.getAs[String]("IMGURL")
        val str49 = row.getAs[String]("PROPERTYPERSONMOBILE")
        val str50 = row.getAs[String]("PROPERTYPERSONTEL")
        val str51 = DataEncry.changPhone(row.getAs[String]("CERTIFICATENUMBE"))
        val str52 = row.getAs[String]("CERTIFICATETYPE")
        val str53 = row.getAs[String]("SIMPLEPINYIN")
        val str54 = row.getAs[String]("FULLPINYIN")
        val str55 = DataEncry.changName(row.getAs[String]("NAME"))
        val str56 = row.getAs[String]("PROPERTYTYPES")
        val str57 = row.getAs[String]("LANDRIGHTSNUMBEDATE")
        val str58 = row.getAs[String]("LANDRIGHTSNUMBE")
        val str59 = DataEncry.changAES(row.getAs[String]("MOBILENUMBER"))
        val str60 = DataEncry.changAES(row.getAs[String]("TELEPHONE"))
        val str61 = DataEncry.changAES(row.getAs[String]("HOUSEOWNERIDCARDNO"))
        val str62 = DataEncry.changAES(row.getAs[String]("CERTIFICATENUMBE"))
        val schema: GenericRowWithSchema = new GenericRowWithSchema(Array(
          str1,
          str2,
          str3,
          str4,
          str5,
          str6,
          str7,
          str8,
          str9,
          str10,
          str11,
          str12,
          str13,
          str14,
          str15,
          str16,
          str17,
          str18,
          str19,
          str20,
          str21,
          str22,
          str23,
          str24,
          str25,
          str26,
          str27,
          str28,
          str29,
          str30,
          str31,
          str32,
          str33,
          str34,
          str35,
          str36,
          str37,
          str38,
          str39,
          str40,
          str41,
          str42,
          str43,
          str44,
          str45,
          str46,
          str47,
          str48,
          str49,
          str50,
          str51,
          str52,
          str53,
          str54,
          str55,
          str56,
          str57,
          str58,
          str59,
          str60,
          str61,
          str62
        ), resSchema)
        list.append(schema)
      }
      list.iterator
    })

    val dataFrame: DataFrame = sparkSession.createDataFrame(value.coalesce(1).cache(), resSchema)

    val path = "/kkk/yichun/zz21_vw_houseinfo/"

    dataFrame.show(false)
    val str: String = dataFrame.toJSON.collectAsList.mkString("[", ",", "]")
    println("执行方法")
    getResult.getData(str, path)

    println("输出完成！")
    sparkSession.close()

  }

}
